package producer

import (
	"errors"
	"fmt"
	"gamesdk/common/log"
	sysTime "gamesdk/common/time"
	"net"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
)

var (
	g_producer  sarama.SyncProducer = nil
	g_ips       []string            = nil
	g_waitGroup sync.WaitGroup
)

//连接mango
func InitProducer(ips []string) {
	defer Trace()()
	if g_producer != nil {
		return
	}
	config := sarama.NewConfig()
	//config.Admin.Timeout = 1 * time.Second
	//config.Net.DialTimeout = 1 * time.Second
	//config.Net.ReadTimeout = 100 * time.Millisecond
	//config.Net.WriteTimeout = 100 * time.Millisecond

	config.Producer.RequiredAcks = sarama.WaitForLocal
	//config.Producer.Timeout = 1 * time.Second
	config.Producer.Flush.Frequency = 500 * time.Millisecond
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true

	producer, err := sarama.NewSyncProducer(ips, config)
	if err != nil {
		panic(err.Error())
	}
	g_producer = producer
	g_ips = ips
	log.Info("connect producer kafka success %v", ips)
}

func SendMsg(topic string, body []byte, errRecv chan error) {
	defer Trace()()
	if g_producer == nil {
		if g_ips != nil {
			InitProducer(g_ips)
		} else {
			panic("please connect mango")
		}
	}
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Partition = int32(-1)
	msg.Key = sarama.StringEncoder("key")
	msg.Value = sarama.ByteEncoder(body)
	g_waitGroup.Add(1)
	go func() {
		_, _, err := g_producer.SendMessage(msg)
		if err != nil {
			//log.Error(err.Error())
			if errRecv != nil {
				errRecv <- err
			}
		} else {
			//log.Debug("send ok partition:%d,offset:%d", partition, offset)
			if errRecv != nil {
				errRecv <- nil
			}
		}
		g_waitGroup.Done()
	}()

}

func SyncSendMsg(topic string, body []byte) error {
	defer Trace()()
	if g_producer == nil {
		if g_ips != nil {
			InitProducer(g_ips)
		} else {
			panic("please connect kafka")
		}
	}
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Partition = int32(-1)
	msg.Key = sarama.StringEncoder("key")
	msg.Value = sarama.ByteEncoder(body)
	_, _, err := g_producer.SendMessage(msg)
	if err != nil {
		return err
	}
	return nil
}

//计算函数所用时间
func Trace() func() {
	fname := ""
	pc, path, line, ok := runtime.Caller(1) // 去掉两层，当前函数和日志的接口函数
	if ok {
		if f := runtime.FuncForPC(pc); f != nil {
			fname = f.Name()
		}
	}
	funcName := lastFname(fname)
	path = getFilePath(path)

	start := time.Now()
	return func() {
		t := sysTime.CurTimeMsStr()
		end := time.Since(start).Nanoseconds() / 1000000.00
		if end >= 100 {
			fmt.Printf("%c[%d;%dm%s %s %s %d (%dms)%c[0m\n", 0x1B, 0, 36, t, path, funcName, line, end, 0x1B)
		}

	}
}

func SyncSendMsgs(topic string, body []string) error {
	defer Trace()()
	if g_producer == nil {
		if g_ips != nil {
			InitProducer(g_ips)
		} else {
			panic("please connect mango")
		}
	}
	var msgs []*sarama.ProducerMessage
	for _, buff := range body {
		msg := &sarama.ProducerMessage{}
		msg.Topic = topic
		msg.Partition = int32(-1)
		msg.Key = sarama.StringEncoder("key")
		msg.Value = sarama.ByteEncoder(buff)
		msgs = append(msgs, msg)
	}
	return g_producer.SendMessages(msgs)
}

//关掉
func Close() {
	if g_producer != nil {
		g_waitGroup.Wait()
		g_producer.Close()
		g_producer = nil
	}
}

func lastFname(fname string) string {
	flen := len(fname)
	n := strings.LastIndex(fname, ".")
	if n+1 < flen {
		return fname[n+1:]
	}
	return fname
}

func getFilePath(path string) string {
	s := strings.Split(path, "src")
	return s[1]
}

func externalIP() (net.IP, error) {
	ifaces, err := net.Interfaces()
	if err != nil {
		return nil, err
	}
	for _, iface := range ifaces {
		if iface.Flags&net.FlagUp == 0 {
			continue // interface down
		}
		if iface.Flags&net.FlagLoopback != 0 {
			continue // loopback interface
		}
		addrs, err := iface.Addrs()
		if err != nil {
			return nil, err
		}
		for _, addr := range addrs {
			ip := getIpFromAddr(addr)
			if ip == nil {
				continue
			}
			return ip, nil
		}
	}
	return nil, errors.New("connected to the network?")
}

func getIpFromAddr(addr net.Addr) net.IP {
	var ip net.IP
	switch v := addr.(type) {
	case *net.IPNet:
		ip = v.IP
	case *net.IPAddr:
		ip = v.IP
	}
	if ip == nil || ip.IsLoopback() {
		return nil
	}
	ip = ip.To4()
	if ip == nil {
		return nil // not an ipv4 address
	}

	return ip
}
